/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.bff.gaia.client.program;

import net.sf.json.JSONException;
import net.sf.json.JSONObject;
import com.bff.gaia.api.common.ExecutionConfig;
import com.bff.gaia.api.common.JobExecutionResult;
import com.bff.gaia.api.common.JobID;
import com.bff.gaia.api.common.Plan;
import com.bff.gaia.api.java.ExecutionEnvironment;
import com.bff.gaia.core.AbstractCompensationFunc;
import com.bff.gaia.client.cli.ServerHandlers;
import com.bff.gaia.client.combine.MergePlanClass;
import com.bff.gaia.client.combine.Qplan;
import com.bff.gaia.optimizer.plan.OptimizedPlan;
import com.bff.gaia.optimizer.plandump.PlanJSONDumpGenerator;
import com.bff.gaia.runtime.jobgraph.SavepointRestoreSettings;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.config.Registry;
import org.apache.http.config.RegistryBuilder;
import org.apache.http.conn.socket.ConnectionSocketFactory;
import org.apache.http.conn.socket.PlainConnectionSocketFactory;
import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
import org.apache.http.conn.ssl.TrustSelfSignedStrategy;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.util.EntityUtils;
import org.json.JSONArray;

import java.io.IOException;
import java.net.URL;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.Collections;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;

/**
 * Execution Environment for remote execution with the Client.
 */
public class ContextEnvironment extends ExecutionEnvironment {

	protected final ClusterClient<?> client;

	protected final List<URL> jarFilesToAttach;

	protected final List<URL> classpathsToAttach;

	protected final ClassLoader userCodeClassLoader;

	protected final SavepointRestoreSettings savepointSettings;
	public static final String CHARSET_UTF_8 = "utf-8";
	// HTTP内容类型。
	public static final String CONTENT_TYPE_TEXT_HTML = "text/xml";
	// HTTP内容类型。相当于form表单的形式，提交数据
	public static final String CONTENT_TYPE_FORM_URL = "application/x-www-form-urlencoded";
	private static PoolingHttpClientConnectionManager pool;
	// 请求配置
	private static RequestConfig requestConfig;
	// HTTP内容类型。相当于form表单的形式，提交数据
	public static final String CONTENT_TYPE_JSON_URL = "application/json;charset=utf-8";
	static {
		try {
			//System.out.println("初始化HttpClientTest~~~开始");
			SSLContextBuilder builder = new SSLContextBuilder();
			builder.loadTrustMaterial(null, new TrustSelfSignedStrategy());
			SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(
				builder.build());
			// 配置同时支持 HTTP 和 HTPPS
			Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory> create().register(
				"http", PlainConnectionSocketFactory.getSocketFactory()).register(
				"https", sslsf).build();
			// 初始化连接管理器
			pool = new PoolingHttpClientConnectionManager(
				socketFactoryRegistry);
			// 将最大连接数增加到200，实际项目最好从配置文件中读取这个值
			pool.setMaxTotal(200);
			// 设置最大路由
			pool.setDefaultMaxPerRoute(2);
			// 根据默认超时限制初始化requestConfig
			int socketTimeout = 10000;
			int connectTimeout = 10000;
			int connectionRequestTimeout = 10000;
			requestConfig = RequestConfig.custom().setConnectionRequestTimeout(
				connectionRequestTimeout).setSocketTimeout(socketTimeout).setConnectTimeout(
				connectTimeout).build();
			//System.out.println("初始化HttpClientTest~~~结束");
		} catch (NoSuchAlgorithmException e) {
			e.printStackTrace();
		} catch (KeyStoreException e) {
			e.printStackTrace();
		} catch (KeyManagementException e) {
			e.printStackTrace();
		}
		// 设置请求超时时间
		requestConfig = RequestConfig.custom().setSocketTimeout(50000).setConnectTimeout(50000)
			.setConnectionRequestTimeout(50000).build();
	}


	public ContextEnvironment(ClusterClient<?> remoteConnection, List<URL> jarFiles, List<URL> classpaths,
				ClassLoader userCodeClassLoader, SavepointRestoreSettings savepointSettings) {
		this.client = remoteConnection;
		this.jarFilesToAttach = jarFiles;
		this.classpathsToAttach = classpaths;
		this.userCodeClassLoader = userCodeClassLoader;
		this.savepointSettings = savepointSettings;
	}

	@Override
	public JobExecutionResult execute(String jobName) throws Exception {
		System.out.println("在ContextEnvironment类中的execute方法体内");
		Plan p = createProgramPlan(jobName);
		ServerHandlers.planList.add(p);
		System.out.println("当前的Plan列表的大小为："+ServerHandlers.planList.size());
		Collections.sort(ServerHandlers.planList, new Comparator<Plan>() {
			@Override
			public int compare(Plan o1, Plan o2) {
				if(o1.getDefaultParallelism()<o2.getDefaultParallelism())
				{
					return -1;
				}
				else if(o1.getDefaultParallelism()==o2.getDefaultParallelism())
				{
					return 0;
				}
				else
				{
					return 1;
				}
			}
		});
		JobWithJars toRun = new JobWithJars(p, this.jarFilesToAttach, this.classpathsToAttach,
			this.userCodeClassLoader);
		ServerHandlers.jarList.add(toRun);
		//当启动的合并的逻辑的时候，就将第一个的返回结果置为空
		if(ServerHandlers.planList.size()==ServerHandlers.ts.size() && ServerHandlers.isRun) {
			System.out.println("作业合并的数目达到了标准，可以进行合并，首先将标记置为true");
			ServerHandlers.isCombine=true;
			if(ServerHandlers.planList.size()==ServerHandlers.args.size() && ServerHandlers.isCombine) {
				System.out.println("准备运行多作业的结果-----》");
				List<Plan> exePlan = new LinkedList<Plan>();
				MergePlanClass MergePC=new MergePlanClass(ServerHandlers.planList,exePlan);
				exePlan=MergePC.getMergePlan();
				System.out.println("exePlan队列大小在函数返回时值为："+exePlan.size());
				ServerHandlers.exePlan.addAll(exePlan);
				List<Qplan> mergeInits=MergePC.getMergeInit();
				System.out.println("MergeInits 队列的大小为："+mergeInits.size());
				List<Plan> todo=MergePC.getPreSub();
				for(int k=0; k<mergeInits.size(); k++){
					exePlan.add(mergeInits.get(k).getPlan());
				}
				System.out.println("全局变量exePlan队列的大小为："+ServerHandlers.exePlan.size());
				//convertCdyToWzm(exePlan);
				for(int j=0;j<exePlan.size(); j++){
					System.out.println(exePlan.get(j).getJobName());
				}
				if(ServerHandlers.jarList.size()>0){
					System.out.println("JarList中有值------------》");
				}
				else{
					System.out.println("当前JarList中的值为空");
				}
				while(ServerHandlers.count < ServerHandlers.exePlan.size()){
					this.lastJobExecutionResult = client.run(toRun, getParallelism(), savepointSettings).getJobExecutionResult();
					System.out.println("得到了一组多作业的执行结果----》,但是当前还没有执行完队列里面的Plan计划");
					ServerHandlers.count++;
					if(this.lastJobExecutionResult==null && ServerHandlers.planList.size() >1)
					{
						throw new Exception("返回结果为空异常");
					}
					else if(ServerHandlers.planList.size()==1){
						this.lastJobExecutionResult = client.run(toRun, getParallelism(), savepointSettings).getJobExecutionResult();
					}
				}
				ServerHandlers.count=0;
			}
			return null;
		}

		else if(!ServerHandlers.isRun )
		{
			this.lastJobExecutionResult = client.run(toRun, getParallelism(), savepointSettings).getJobExecutionResult();
			if(this.lastJobExecutionResult==null )
			{
				throw new Exception("返回结果为空异常");
			}
		}
		else if(ServerHandlers.planList.size()==ServerHandlers.args.size() && ServerHandlers.isCombine)
		{
			System.out.println("准备运行多作业的结果-----》");
			while(ServerHandlers.planList.size()!=0){
				String str="http://localhost:8081/taskmanagers";
				int jm=get(str);
				while(jm<0){
					jm=get(str);
				}
				System.out.println("当前作业管理器中的可用资源大小为："+jm);
				if(jm >= ServerHandlers.planList.get(0).getMaximumParallelism()) {
					System.out.println("当前资源管理器的可用slot数量满足条件");
					this.lastJobExecutionResult = client.run(toRun, getParallelism(), savepointSettings).getJobExecutionResult();
					System.out.println("得到了一组多作业的执行结果----》");
				}

				if(this.lastJobExecutionResult==null && ServerHandlers.planList.size() >= 2)
				{
					throw new Exception("返回结果为空异常");
				}
			}


		}
		//System.out.println("这次提交的作业执行经过contextEnvironment环境的execute方法");

		return this.lastJobExecutionResult;
	}
	public static CloseableHttpClient getHttpClient() {

		CloseableHttpClient httpClient = HttpClients.custom()
			// 设置连接池管理
			.setConnectionManager(pool)
			// 设置请求配置
			.setDefaultRequestConfig(requestConfig)
			// 设置重试次数
			.setRetryHandler(new DefaultHttpRequestRetryHandler(0, false))
			.build();

		return httpClient;
	}
	private static String  sendHttpGet(HttpGet httpGet){
		CloseableHttpClient httpClient = null;
		CloseableHttpResponse response = null;
		// 响应内容
		String responseContent = null;
		try {
			httpClient = getHttpClient();
			httpGet.setConfig(requestConfig);
			response = httpClient.execute(httpGet);
			HttpEntity entity = response.getEntity();
			if (response.getStatusLine().getStatusCode() >= 300) {
				throw new Exception(
					"HTTP Request is not success, Response code is " + response.getStatusLine().getStatusCode());
			}
			if (HttpStatus.SC_OK == response.getStatusLine().getStatusCode()) {
				responseContent = EntityUtils.toString(entity, CHARSET_UTF_8);
				EntityUtils.consume(entity);
			}

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				// 释放资源
				if (response != null) {
					response.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return responseContent;
	}
	public static String HttpGet(String httpUrl){
		//创建Get请求
		HttpGet httpGet=new HttpGet(httpUrl);
		return sendHttpGet(httpGet);
	}
	public static int get(String s){
		try {
			String tempdata=HttpGet(s);
			JSONObject pa=JSONObject.fromObject(tempdata);
			String accountNo=pa.getString("taskmanagers");
			JSONArray jArray = new JSONArray(accountNo);
			JSONObject ps=JSONObject.fromObject(jArray.get(0).toString());
			String ss=ps.getString("freeSlots");
			int kk=Integer.parseInt(ss);
			return kk;

		} catch (JSONException e){
			e.printStackTrace();
		} catch(NullPointerException e){
			e.printStackTrace();
		}
		return 0;
	}

	//task2.3_iteration
	public JobExecutionResult execute(String jobName, AbstractCompensationFunc compensationFunc) throws Exception {
		Plan p = createProgramPlan(jobName);
		p.setCompensationFunc(compensationFunc);
		JobWithJars toRun = new JobWithJars(p, this.jarFilesToAttach, this.classpathsToAttach,
			this.userCodeClassLoader);
		this.lastJobExecutionResult = client.run(toRun, getParallelism(), savepointSettings).getJobExecutionResult();
		return this.lastJobExecutionResult;
	}
	//end

	@Override
	public String getExecutionPlan() throws Exception {
		Plan plan = createProgramPlan("unnamed job");

		OptimizedPlan op = ClusterClient.getOptimizedPlan(client.compiler, plan, getParallelism());
		PlanJSONDumpGenerator gen = new PlanJSONDumpGenerator();
		return gen.getOptimizerPlanAsJSON(op);
	}

	@Override
	public void startNewSession() throws Exception {
		client.endSession(jobID);
		jobID = JobID.generate();
	}

	@Override
	public String toString() {
		return "Context Environment (parallelism = " + (getParallelism() == ExecutionConfig.PARALLELISM_DEFAULT ? "default" : getParallelism())
				+ ") : " + getIdString();
	}

	public ClusterClient<?> getClient() {
		return this.client;
	}

	public List<URL> getJars(){
		return jarFilesToAttach;
	}

	public List<URL> getClasspaths(){
		return classpathsToAttach;
	}

	public ClassLoader getUserCodeClassLoader() {
		return userCodeClassLoader;
	}

	public SavepointRestoreSettings getSavepointRestoreSettings() {
		return savepointSettings;
	}

	// --------------------------------------------------------------------------------------------

	static void setAsContext(ContextEnvironmentFactory factory) {
		initializeContextEnvironment(factory);
	}

	static void unsetContext() {
		resetContextEnvironment();
	}
}